﻿using System;
using System.Timers;
using Communication.Lib.Core;
using Communication.Lib.Mode;
using Confluent.Kafka;
using log4net;

namespace Communication.Lib.Communication
{
    public class KafkaCommunicate<T> : ICommunicatable<T> where T : KafkaBaseInfo
    {
        private ILog logger = LogManager.GetLogger(typeof(KafkaCommunicate<T>));

        private TopicName topicName;
        private ConsumerConfig confC;
        private ProducerConfig _confP;
        private IConsumer<Ignore, string> consumer;
        private IProducer<Null, string> producer;
        private Timer timerListen;
        private DateTime _lastDateTime;
        private Action<DeliveryReport<Null, string>> handlerP;

        public event Action<T> RecvHandler;

        public bool IsAlive => (DateTime.Now - _lastDateTime).TotalSeconds < 30;

        /// <summary>
        /// 默认与Consumer的BootstrapServers一致
        /// </summary>
        public ProducerConfig ConfP
        {
            get { return _confP; }
            set
            {
                if (value != _confP)
                {
                    _confP = value;
                    ReInitProducer();
                }
            }
        }

        public KafkaCommunicate(string groupid, string host, TopicName topic)
        {
            topicName = topic;

            confC = new ConsumerConfig()
            {
                GroupId = groupid,
                BootstrapServers = host,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = true,
                FetchWaitMaxMs = 50,
            };

            _confP = new ProducerConfig()
            {
                BootstrapServers = host
            };
            consumer = new ConsumerBuilder<Ignore, string>(confC).SetErrorHandler(ErrorHandler).Build();
            consumer.Subscribe(topicName.ToString());

            producer = new ProducerBuilder<Null, string>(ConfP).Build();
            handlerP = r =>
            {
                if (r.Error.IsError)
                {
                    logger.Error($"send KafkaDataError:{r.Value};{r.Error.Reason}");
                }
            };

            timerListen = new Timer();
            timerListen.Interval = 200;
            timerListen.Elapsed += TimerListen_Elapsed;
        }

        private void ReInitProducer()
        {
            producer = null;
            producer = new ProducerBuilder<Null, string>(ConfP).Build();
        }

        private void TimerListen_Elapsed(object sender, ElapsedEventArgs e)
        {
            try
            {
                var cc = consumer.Consume(TimeSpan.FromSeconds(1));
                if (!string.IsNullOrEmpty(cc?.Value))
                {
                    _lastDateTime = DateTime.Now;

                    T t = Activator.CreateInstance<T>();
                    t.TopicName = topicName;
                    t.Data = cc.Value;
                    t.Handle();
                    RecvHandler?.Invoke(t);
                }
            }
            catch (Exception ex)
            {
                logger.Error($"get kafkaData Error:TopicName  {topicName.ToString()},message :{ex}");
            }
        }

        private void ErrorHandler(IConsumer<Ignore, string> c, Error e)
        {
            if (c.Subscription.Count > 0)
            {
                logger.Error($"topicName:{c.Subscription[0]};error:{e.Reason}");
                var consumer = new ConsumerBuilder<Ignore, string>(confC).SetErrorHandler(ErrorHandler).Build();
                consumer.Subscribe(c.Subscription[0]);
            }
            else { logger.Error($"UnKnonwError : {e.Reason}"); }
        }

        public bool SendData(T info)
        {
            if (producer == null)
            {
                ReInitProducer();
            }

            try
            {
                producer?.Produce(info.TopicName.ToString(), new Message<Null, string>() { Value = info.Data }, handlerP);
                producer?.Flush(TimeSpan.FromSeconds(1));
            }
            catch (Exception e)
            {
                logger.Error($"sendData Error.topicName :{info.TopicName.ToString()},data:{info.Data}.\r\n{e}");
                return false;
            }
            return true;
        }

        public virtual bool SendData(byte[] bytes)
        {
            return true;
        }

        public void Start()
        {
            timerListen.Start();
        }

        public void Stop()
        {
            timerListen.Stop();
        }
    }
}